package org.jpush.core;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

public abstract class AbstractChannelStreamer
{
	public static enum ProcessResult
	{
		DeleteItem,
		PauseProcessing,
		ContinueProcessing
	};
	
	public static enum SendResult
	{
	    SendResult_OK,
	    SendResult_Retry,
	    SendResult_NotOK
	};

    private volatile int nConnectionCount;

    private List<LogicalConnection> activeConnectionList;
    private List<LogicalConnection> inactiveConnectionList;
    private List<LogicalConnection> newAddedConnectionList;

    private AtomicLong lastQueueChange;

    private long lastInactiveListCheckTime;
    private long lastAddedConnectionTime;
    private long lastAddedConnectionSyncTime;

    private boolean stop;

    private Thread thread;

	public AbstractChannelStreamer()
	{
	    stop = false;
	    activeConnectionList = new LinkedList<LogicalConnection>();
	    inactiveConnectionList = new LinkedList<LogicalConnection>();
	    newAddedConnectionList = new LinkedList<LogicalConnection>();

	    lastQueueChange = new AtomicLong(0);
	    lastAddedConnectionTime = 0;
	    lastInactiveListCheckTime = System.currentTimeMillis();
	    lastAddedConnectionSyncTime = System.currentTimeMillis();
	}

    public synchronized void reshuffle()
    {
    	lastQueueChange.getAndSet(System.currentTimeMillis());
    }

    public void addItem(LogicalConnection pItem)
    {
    	synchronized (newAddedConnectionList)
    	{
    		this.newAddedConnectionList.add(pItem);
    		this.nConnectionCount = this.nConnectionCount + 1;
    		this.lastAddedConnectionTime = System.currentTimeMillis();
    	}
    }

    public int getItemsCount()
    {
        return nConnectionCount;
    }

    public void start()
    {
    	final AbstractChannelStreamer _self = this;
    	thread = new Thread (new Runnable()
    	{
			public void run()
			{
				_self.doWork();
			}
    		
    	});
    }

    public void stop()
    {
        stop = true;
        try
        {
			thread.join();
		}
        catch (Exception e)
		{
			e.printStackTrace();
		}
    }

    /**
     * Check if there are new added connections, if found
     * move them into active list
     * 
     * @return true if found
     */
    private boolean checkMergeNewAddedConnectionList()
    {
        boolean ret = false;
        // There is new added connection but not synchronized
        if (lastAddedConnectionTime > lastAddedConnectionSyncTime)
        {
            synchronized(newAddedConnectionList)
            {
            	// Move new added connections into active connection list
                if (!newAddedConnectionList.isEmpty())
                {
                    ret = true;
                    activeConnectionList.addAll(newAddedConnectionList);

                    newAddedConnectionList.clear();
                }

                lastAddedConnectionSyncTime = System.currentTimeMillis();
            }
        }
        return ret;
    }


    /**
     * Check if there are inactive connections, if found
     * move them into active list
     * 
     * @return true if found
     */
    private boolean checkMergeInactiveConnectionList()
    {
        boolean ret = false;
        // 
        if (lastQueueChange.get() > lastInactiveListCheckTime)
        {
        	// Move inactive connections into active connection list
            if (!inactiveConnectionList.isEmpty())
            {
                ret = true;
                activeConnectionList.addAll(inactiveConnectionList);

                inactiveConnectionList.clear();
            }
            lastInactiveListCheckTime = System.currentTimeMillis();
        }

        return ret;
    }

    private void processActiveConnectionList()
    {
    	LogicalConnection logicalConnection = null;
    	ProcessResult result = null;
    	
    	Iterator<LogicalConnection> it = activeConnectionList.iterator();
    	while (it.hasNext())
    	{
        	logicalConnection = it.next();
        	result = processConnection(logicalConnection);

            if(result == ProcessResult.DeleteItem)
            {
            	it.remove();
                this.nConnectionCount = this.nConnectionCount - 1;
            }
            else if (result == ProcessResult.PauseProcessing)
            {
                inactiveConnectionList.add(logicalConnection);
            	it.remove();
                lastInactiveListCheckTime = System.currentTimeMillis();
            }
            else if (result == ProcessResult.ContinueProcessing)
            {
            	// nothing will be done.
            }

            // Reach to the end, check and merge
            // and do again until all the active connection 
            // have been processed
            if (!it.hasNext())
            {
                checkMergeNewAddedConnectionList();
                checkMergeInactiveConnectionList();
                it = activeConnectionList.iterator();
            }
    	}
    }

    private void doWork()
    {
    	while (!stop)
        {
            if (checkMergeNewAddedConnectionList() || checkMergeInactiveConnectionList() )
            {
                processActiveConnectionList();
            }
            else
            {
                try
                {
					Thread.sleep(1000);
				}
                catch (Exception e)
                {
					e.printStackTrace();
				}
            }
        }
    }

    protected abstract ProcessResult processConnection(LogicalConnection pItem);
}
